package streamproc.filtercomponents;

import java.sql.Timestamp;
import java.util.HashMap;
import java.util.Vector;

import event.Event;
import event.EventFormatManager;
import event.DEFINES.CEPLEventDataFieldType;
import event.UnknownDataFieldException;

import streamdef.StreamCondition;
import streamdef.StreamDefinitionManager;
import streamdef.EventSet;
import streamdef.ConditionStatement;
import streamdef.ConditionStatementFieldFromOneStream;
import streamdef.ConditionStatementFieldFromTwoStreams;
import streamdef.ConditionStatementTimeRelationTwoStreams;
import streamdef.SubCondition;
import streamdef.DEFINES.FieldRelationType;
import streamdef.DEFINES.ConditionStatementType;
import streamproc.ChainComponent;
import streamproc.buffercomponents.Buffer;
import streamproc.buffercomponents.BufferManagerComponent;

public class EventSetFilterComponent extends ChainComponent {
	private static EventSetFilterComponent sInstance = null;

	public static EventSetFilterComponent instance() {
		if (sInstance == null)
			sInstance = new EventSetFilterComponent();

		return sInstance;
	}

	private HashMap<Long, StreamCondition> streamConditions = null;

	private EventSetFilterComponent() {
		streamConditions = new HashMap<Long, StreamCondition>();
	}

	public void reset() {
		if (streamConditions != null)
			streamConditions.clear();
	}

	protected void finalize() {
		reset();
	}

	public void init() {
		reset();

		Vector<Long> queryIDs = StreamDefinitionManager.instance()
				.getStreamIDs();
		for (int i = 0; i < queryIDs.size(); ++i) {
			long qID = queryIDs.get(i);
			StreamCondition mq = StreamDefinitionManager.instance()
					.getParsedQuery(qID);
			streamConditions.put(new Long(qID), mq);
		}
	}

	// each queryresult is a valid sequence
	// -> BEFORE-statements don't need to be checked anymore
	// -> comparison statements and non-occurences must be checked !
	public synchronized void processQueryResults(Vector<EventSet> v) {
		Vector<EventSet> result = new Vector<EventSet>();

		for (int i = 0; i < v.size(); ++i) {
			EventSet temp = v.get(i);
			StreamCondition mq = streamConditions.get(temp.getStreamID());

			if (isValidSequenceForMainQuery(temp, mq))
				result.add(temp);
		}

		v.clear();
		next.processQueryResults(result);
	}

	private boolean isValidSequenceForMainQuery(EventSet qr, StreamCondition mq) {
		Vector<Long> forbiddenEvents = mq
				.getInternalStreamIDsNotInStatementAndNonOccurring();
		for (int i = 0; i < forbiddenEvents.size(); ++i) {
			double range = mq.getRangeForInternalStreamID(forbiddenEvents
					.get(i));
			long extID = mq.getExternalStreamID(forbiddenEvents.get(i));
			Buffer b = BufferManagerComponent.instance().getBuffer(extID);

			long ts = qr.getMostRecentTimestamp().getTime();
			ts -= (1000 * range);
			Timestamp start = new Timestamp(ts);

			if (b.eventExists(start, null, qr.getEventIDs()) != null)
				return false;
		}

		Vector<SubCondition> subConditions = mq.getSubQueries();
		for (int i = 0; i < subConditions.size(); ++i) {
			SubCondition sq = subConditions.get(i);
			try {
				if (!isValidSequenceForSubQuery(qr, mq, sq))
					return false;
			} catch (UnknownDataFieldException e) {
				e.printStackTrace();
			}
		}

		return true;
	}

	private boolean isValidSequenceForSubQuery(EventSet qr, StreamCondition mq,
			SubCondition sq) throws UnknownDataFieldException {
		boolean isAnd = sq.isANDStatement();
		Vector<ConditionStatement> stats = sq.getQueryStatements();

		boolean result = true;
		if (!isAnd)
			result = false;
		Vector<Long> intIDs = sq.getUsedInternalStreamIDs();

		// may all occur
		if (mq.mayOccur(intIDs.get(0))
				&& ((intIDs.size() > 1 && mq.mayOccur(intIDs.get(1))) || intIDs
						.size() == 1)) {
			for (int i = 0; i < stats.size(); ++i) {
				ConditionStatement qs = stats.get(i);
				boolean isValid = isValidSequenceForQueryStatement(qr, mq, sq,
						qs);

				if (isValid && !isAnd)
					result = true;
				if (!isValid && isAnd)
					result = false;
			}
		}
		// one forbidden stream
		else {
			if (!isAnd)
				return false;

			Vector<Long> eventIDsToExclude = qr.getEventIDs();
			Vector<Integer> indicesToCheck = null;
			Vector<Integer> bufferLimits = null;
			Long forbiddenStreamID = mq.mayOccur(intIDs.get(0)) ? intIDs.get(1)
					: intIDs.get(0);
			Long allowedStreamID = mq.mayOccur(intIDs.get(0)) ? intIDs.get(0)
					: intIDs.get(1);

			// get the BEFORE-statement if available
			ConditionStatementTimeRelationTwoStreams stat = null;
			for (int i = 0; i < stats.size(); ++i) {
				ConditionStatement qs = stats.get(i);
				if (qs.getType() == ConditionStatementType.TIMERELATION_TWOSTREAMS)
					stat = (ConditionStatementTimeRelationTwoStreams) qs;
			}
			if (stat != null)
				bufferLimits = isValidForQueryStatementTimeNotOccurringInBuffer(
						qr, mq, sq, stat);
			else {
				// it is not available
				double range = mq
						.getRangeForInternalStreamID(forbiddenStreamID);
				long extID = mq.getExternalStreamID(forbiddenStreamID);
				Buffer b = BufferManagerComponent.instance().getBuffer(extID);

				long ts = qr.getMostRecentTimestamp().getTime();
				ts -= (1000 * range);
				Timestamp start = new Timestamp(ts);

				bufferLimits = b.eventExists(start,
						qr.getMostRecentTimestamp(), eventIDsToExclude);
			}

			if (bufferLimits == null || bufferLimits.size() == 0)
				return true;

			for (int i = 0; i < stats.size(); ++i) {
				ConditionStatement qs = stats.get(i);
				if (qs.getType() == ConditionStatementType.FIELD_ONESTREAM) {
					ConditionStatementFieldFromOneStream qss = (ConditionStatementFieldFromOneStream) qs;
					if (allowedStreamID == qss.getInternalStreamID()) {
						if (!isValidSequenceForQueryStatementOneStream(qr, mq,
								sq, qss))
							return false;
					} else {
						indicesToCheck = isValidSequenceForQueryStatementOneStreamNotOccuringInBuffer(
								qr, mq, sq, qss, indicesToCheck, bufferLimits);
						if (indicesToCheck == null
								|| indicesToCheck.size() == 0)
							return true;
					}
				} else if (qs.getType() == ConditionStatementType.FIELD_TWOSTREAMS) {
					ConditionStatementFieldFromTwoStreams qss = (ConditionStatementFieldFromTwoStreams) qs;
					indicesToCheck = isValidSequenceForQueryStatementTwoStreamsNotOccuringInBuffer(
							qr, mq, sq, qss, indicesToCheck, bufferLimits);
					if (indicesToCheck == null || indicesToCheck.size() == 0)
						return true;
				}
			}

			return false;
		}

		return result;
	}

	private boolean isValidSequenceForQueryStatement(EventSet qr,
			StreamCondition mq, SubCondition sq, ConditionStatement qs) {
		try {
			if (qs.getType() == ConditionStatementType.FIELD_ONESTREAM) {
				ConditionStatementFieldFromOneStream qsOs = (ConditionStatementFieldFromOneStream) qs;
				return isValidSequenceForQueryStatementOneStream(qr, mq, sq,
						qsOs);
			} else if (qs.getType() == ConditionStatementType.FIELD_TWOSTREAMS) {
				ConditionStatementFieldFromTwoStreams qsOs = (ConditionStatementFieldFromTwoStreams) qs;
				return isValidSequenceForQueryStatementTwoStreams(qr, mq, sq,
						qsOs);
			} else if (qs.getType() == ConditionStatementType.TIMERELATION_TWOSTREAMS) {
				ConditionStatementTimeRelationTwoStreams qsOs = (ConditionStatementTimeRelationTwoStreams) qs;
				return isValidSequenceForQueryStatementTime(qr, mq, sq, qsOs);
			}

			return false;
		} catch (UnknownDataFieldException e) {
			System.err.println(e);
			return false;
		}
	}

	private boolean isValidSequenceForQueryStatementOneStream(EventSet qr,
			StreamCondition mq, SubCondition sq,
			ConditionStatementFieldFromOneStream qs)
			throws UnknownDataFieldException {
		boolean result = false;

		Event e = qr.getEventForInternalStreamID(qs.getInternalStreamID());

		String fieldName = qs.getField();
		CEPLEventDataFieldType type = EventFormatManager.instance()
				.getFieldValueType(e.getStreamID(), fieldName);
		if (type == CEPLEventDataFieldType.TEXT) {
			String valueFromStat = (String) qs.getFieldValue();
			String valueFromEvent = e.getString(fieldName);

			switch (qs.getRelation()) {
			case EQUALS:
				result = valueFromEvent.compareToIgnoreCase(valueFromStat) == 0;
				break;
			case NOTEQUALS:
				result = valueFromEvent.compareToIgnoreCase(valueFromStat) != 0;
				break;
			default:
				return false;
			}
		} else {
			double valueFromStat = 0;
			double valueFromEvent = 0;

			switch (type) {
			case CHAR: {
				Character vStat = (Character) qs.getFieldValue();
				valueFromEvent = e.getCharacter(fieldName);
				valueFromStat = vStat.charValue();
			}
				break;
			case SHORT: {
				Short vStat = (Short) qs.getFieldValue();
				valueFromEvent = e.getShort(fieldName);
				valueFromStat = vStat.shortValue();
			}
				break;
			case INT: {
				Integer vStat = (Integer) qs.getFieldValue();
				valueFromEvent = e.getInteger(fieldName);
				valueFromStat = vStat.intValue();
			}
				break;
			case LONG: {
				Long vStat = (Long) qs.getFieldValue();
				valueFromEvent = e.getLong(fieldName);
				valueFromStat = vStat.longValue();
			}
				break;
			case FLOAT: {
				Float vStat = (Float) qs.getFieldValue();
				valueFromEvent = e.getFloat(fieldName);
				valueFromStat = vStat.floatValue();
			}
				break;

			case DOUBLE: {
				Double vStat = (Double) qs.getFieldValue();
				valueFromEvent = e.getDouble(fieldName);
				valueFromStat = vStat.doubleValue();
			}
				break;
			default:
				return false;
			}

			switch (qs.getRelation()) {
			case EQUALS:
				result = valueFromEvent == valueFromStat;
				break;
			case NOTEQUALS:
				result = valueFromEvent != valueFromStat;
				break;
			case EQUALSORGREATER:
				result = valueFromEvent >= valueFromStat;
				break;
			case EQUALSORLESSER:
				result = valueFromEvent <= valueFromStat;
				break;
			case GREATER:
				result = valueFromEvent > valueFromStat;
				break;
			case LESSER:
				result = valueFromEvent < valueFromStat;
				break;
			default:
				return false;
			}
		}

		return result;
	}

	private boolean isValidSequenceForQueryStatementTwoStreams(EventSet qr,
			StreamCondition mq, SubCondition sq,
			ConditionStatementFieldFromTwoStreams qs)
			throws UnknownDataFieldException {
		boolean result = false;

		Event left = qr.getEventForInternalStreamID(qs.internalStreamID1);
		Event right = qr.getEventForInternalStreamID(qs.internalStreamID2);

		String fieldLeft = qs.field1;
		String fieldRight = qs.field2;

		CEPLEventDataFieldType typeLeft = EventFormatManager.instance()
				.getFieldValueType(left.getStreamID(), fieldLeft);
		CEPLEventDataFieldType typeRight = EventFormatManager.instance()
				.getFieldValueType(right.getStreamID(), fieldRight);

		if (typeLeft == CEPLEventDataFieldType.TEXT
				|| typeRight == CEPLEventDataFieldType.TEXT) {
			String valueFromLeft = null;
			String valueFromRight = null;

			switch (typeLeft) {
			case TEXT:
				valueFromLeft = left.getString(fieldLeft);
				break;
			case CHAR:
				Character tempC = left.getCharacter(fieldLeft);
				valueFromLeft = tempC.toString();
				break;
			case SHORT:
				Short tempS = left.getShort(fieldLeft);
				valueFromLeft = tempS.toString();
				break;
			case INT:
				Integer tempI = left.getInteger(fieldLeft);
				valueFromLeft = tempI.toString();
				break;
			case LONG:
				Long tempL = left.getLong(fieldLeft);
				valueFromLeft = tempL.toString();
				break;
			case FLOAT:
				Float tempF = left.getFloat(fieldLeft);
				valueFromLeft = tempF.toString();
				break;
			case DOUBLE:
				Double tempD = left.getDouble(fieldLeft);
				valueFromLeft = tempD.toString();
				break;
			default:
				return false;
			}

			switch (typeRight) {
			case TEXT:
				valueFromRight = right.getString(fieldRight);
				break;
			case CHAR:
				Character tempC = right.getCharacter(fieldRight);
				valueFromRight = tempC.toString();
				break;
			case SHORT:
				Short tempS = right.getShort(fieldRight);
				valueFromRight = tempS.toString();
				break;
			case INT:
				Integer tempI = right.getInteger(fieldRight);
				valueFromRight = tempI.toString();
				break;
			case LONG:
				Long tempL = right.getLong(fieldRight);
				valueFromRight = tempL.toString();
				break;
			case FLOAT:
				Float tempF = right.getFloat(fieldRight);
				valueFromRight = tempF.toString();
				break;
			case DOUBLE:
				Double tempD = right.getDouble(fieldRight);
				valueFromRight = tempD.toString();
				break;
			default:
				return false;
			}

			switch (qs.relation) {
			case EQUALS:
				result = valueFromLeft.compareToIgnoreCase(valueFromRight) == 0;
				break;
			case NOTEQUALS:
				result = valueFromLeft.compareToIgnoreCase(valueFromRight) != 0;
				break;
			default:
				return false;
			}
		} else {
			double valueFromLeft = 0;
			double valueFromRight = 0;

			switch (typeLeft) {
			case CHAR:
				valueFromLeft = left.getCharacter(fieldLeft);
				break;
			case SHORT:
				valueFromLeft = left.getShort(fieldLeft);
				break;
			case INT:
				valueFromLeft = left.getInteger(fieldLeft);
				break;
			case LONG:
				valueFromLeft = left.getLong(fieldLeft);
				break;
			case FLOAT:
				valueFromLeft = left.getFloat(fieldLeft);
				break;
			case DOUBLE:
				valueFromLeft = left.getDouble(fieldLeft);
				;
				break;
			default:
				return false;
			}

			switch (typeRight) {
			case CHAR:
				valueFromRight = right.getCharacter(fieldRight);
				break;
			case SHORT:
				valueFromRight = right.getShort(fieldRight);
				;
				break;
			case INT:
				valueFromRight = right.getInteger(fieldRight);
				break;
			case LONG:
				valueFromRight = right.getLong(fieldRight);
				break;
			case FLOAT:
				valueFromRight = right.getFloat(fieldRight);
				break;
			case DOUBLE:
				valueFromRight = right.getDouble(fieldRight);
				break;
			default:
				return false;
			}

			switch (qs.relation) {
			case EQUALS:
				result = valueFromLeft == valueFromRight;
				break;
			case NOTEQUALS:
				result = valueFromLeft != valueFromRight;
				break;
			case EQUALSORGREATER:
				result = valueFromLeft >= valueFromRight;
				break;
			case EQUALSORLESSER:
				result = valueFromLeft <= valueFromRight;
				break;
			case GREATER:
				result = valueFromLeft > valueFromRight;
				break;
			case LESSER:
				result = valueFromLeft < valueFromRight;
				break;
			default:
				return false;
			}
		}

		return result;
	}

	private boolean isValidSequenceForQueryStatementTime(EventSet qr,
			StreamCondition mq, SubCondition sq,
			ConditionStatementTimeRelationTwoStreams qs) {
		boolean result = false;

		Event left = qr.getEventForInternalStreamID(qs.internalStreamID1);
		Event right = qr.getEventForInternalStreamID(qs.internalStreamID2);

		result = left.getTimestamp().before(right.getTimestamp());

		return result;
	}

	private Vector<Integer> isValidSequenceForQueryStatementOneStreamNotOccuringInBuffer(
			EventSet qr, StreamCondition mq, SubCondition sq,
			ConditionStatementFieldFromOneStream qs,
			Vector<Integer> indicesToCheck, Vector<Integer> bufferLimits)
			throws UnknownDataFieldException {
		Long extID = mq.getExternalStreamID(qs.getInternalStreamID());
		Buffer b = BufferManagerComponent.instance().getBuffer(extID);
		String field = qs.getField();
		CEPLEventDataFieldType type = EventFormatManager.instance()
				.getFieldValueType(extID, field);

		if (type == CEPLEventDataFieldType.TEXT) {
			String valueFromStat = (String) qs.getFieldValue();
			return b.valueExists(field, qs.relation, valueFromStat, false,
					bufferLimits, indicesToCheck, qr.getEventIDs());
		} else {
			double valueFromStat = 0;

			switch (type) {
			case CHAR: {
				Character vStat = (Character) qs.getFieldValue();
				valueFromStat = vStat.charValue();
			}
				break;
			case SHORT: {
				Short vStat = (Short) qs.getFieldValue();
				valueFromStat = vStat.shortValue();
			}
				break;
			case INT: {
				Integer vStat = (Integer) qs.getFieldValue();
				valueFromStat = vStat.intValue();
			}
				break;
			case LONG: {
				Long vStat = (Long) qs.getFieldValue();
				valueFromStat = vStat.longValue();
			}
				break;
			case FLOAT: {
				Float vStat = (Float) qs.getFieldValue();
				valueFromStat = vStat.floatValue();
			}
				break;

			case DOUBLE: {
				Double vStat = (Double) qs.getFieldValue();
				valueFromStat = vStat.doubleValue();
			}
				break;
			}

			return b.valueExists(field, qs.relation, valueFromStat, true,
					bufferLimits, indicesToCheck, qr.getEventIDs());

		}
	}

	private Vector<Integer> isValidSequenceForQueryStatementTwoStreamsNotOccuringInBuffer(
			EventSet qr, StreamCondition mq, SubCondition sq,
			ConditionStatementFieldFromTwoStreams qs,
			Vector<Integer> indicesToCheck, Vector<Integer> bufferLimits)
			throws UnknownDataFieldException {
		long extIDLeft = mq.getExternalStreamID(qs.getInternalStreamIDLeft());
		long extIDRight = mq.getExternalStreamID(qs.getInternalStreamIDRight());
		String fieldLeft = qs.getFieldLeft();
		String fieldRight = qs.getFieldLeft();

		// right part may not occur
		if (mq.mayOccur(qs.getInternalStreamIDLeft())
				&& !mq.mayOccur(qs.getInternalStreamIDRight())) {
			Buffer b = BufferManagerComponent.instance().getBuffer(extIDRight);

			// invert relation
			FieldRelationType origRel = qs.getRelation();
			FieldRelationType newRel = origRel;
			if (origRel == FieldRelationType.EQUALSORGREATER)
				newRel = FieldRelationType.EQUALSORLESSER;
			else if (origRel == FieldRelationType.GREATER)
				newRel = FieldRelationType.LESSER;
			else if (origRel == FieldRelationType.EQUALSORLESSER)
				newRel = FieldRelationType.EQUALSORGREATER;
			else if (origRel == FieldRelationType.LESSER)
				newRel = FieldRelationType.GREATER;

			Event e = qr.getEventForInternalStreamID(qs
					.getInternalStreamIDLeft());
			Object value = null;
			boolean isDouble = false;

			if (EventFormatManager.instance().getFieldValueType(extIDLeft,
					fieldLeft) == CEPLEventDataFieldType.TEXT)
				try {
					value = (Object) e.getString(fieldLeft);
				} catch (UnknownDataFieldException e1) {
					e1.printStackTrace();
				}
			else {
				value = e.getEncapsulatingDouble(fieldLeft);
				isDouble = true;
			}

			return b.valueExists(fieldRight, newRel, value, isDouble,
					bufferLimits, indicesToCheck, qr.getEventIDs());
		}
		// left part may not occur
		else if (!mq.mayOccur(qs.getInternalStreamIDLeft())
				&& mq.mayOccur(qs.getInternalStreamIDRight())) {
			Buffer b = BufferManagerComponent.instance().getBuffer(extIDLeft);

			// keep relation
			FieldRelationType origRel = qs.getRelation();
			FieldRelationType newRel = origRel;

			Event e = qr.getEventForInternalStreamID(qs
					.getInternalStreamIDRight());
			Object value = null;
			boolean isDouble = false;

			if (EventFormatManager.instance().getFieldValueType(extIDRight,
					fieldRight) == CEPLEventDataFieldType.TEXT)
				try {
					value = e.getString(fieldRight);
				} catch (UnknownDataFieldException e1) {
					e1.printStackTrace();
				}
			else {
				value = e.getEncapsulatingDouble(fieldRight);
				isDouble = true;
			}

			return b.valueExists(fieldRight, newRel, value, isDouble,
					bufferLimits, indicesToCheck, qr.getEventIDs());
		}
		// both parts may not occur -> join between two buffers
		// -> not supported
		else
			return null;
	}

	// gets the border limits of the buffer, if events exist
	private Vector<Integer> isValidForQueryStatementTimeNotOccurringInBuffer(
			EventSet qr, StreamCondition mq, SubCondition sq,
			ConditionStatementTimeRelationTwoStreams qs) {
		long intIDLeft = qs.getInternalStreamIDLeft();
		long intIDRight = qs.getInternalStreamIDRight();
		long extIDLeft = mq.getExternalStreamID(intIDLeft);
		long extIDRight = mq.getExternalStreamID(intIDRight);

		// left part may not occur
		if (!mq.mayOccur(intIDLeft) && mq.mayOccur(intIDRight)) {
			Event e = qr.getEventForInternalStreamID(intIDRight);
			Buffer b = BufferManagerComponent.instance().getBuffer(extIDLeft);

			long ts = qr.getMostRecentTimestamp().getTime();
			ts -= (1000 * mq.getRangeForInternalStreamID(qs
					.getInternalStreamIDRight()));
			Timestamp begin = new Timestamp(ts);
			Timestamp end = e.getTimestamp();

			return b.eventExists(begin, end, qr.getEventIDs());
		}
		// right part may not occur
		else if (mq.mayOccur(intIDLeft) && !mq.mayOccur(intIDRight)) {
			Event e = qr.getEventForInternalStreamID(intIDLeft);
			Buffer b = BufferManagerComponent.instance().getBuffer(extIDRight);

			long ts = qr.getMostRecentTimestamp().getTime();
			ts -= (1000 * mq.getRangeForInternalStreamID(qs
					.getInternalStreamIDRight()));

			Timestamp begin = null;
			if (e.getTimestamp().getTime() < ts)
				begin = new Timestamp(ts);
			else
				begin = e.getTimestamp();

			Timestamp end = null;

			return b.eventExists(begin, end, qr.getEventIDs());
		}
		// both parts may not occur -> join between two buffers
		// -> not supported
		else
			return null;
	}
}
